package com.easylinkin.linkappapi.openapi.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.easylinkin.bases.redis.util.RedisUtil;
import com.easylinkin.linkappapi.airconditioner.service.AirconditionDeviceRecordService;
import com.easylinkin.linkappapi.alarm.service.AiAlarmCountService;
import com.easylinkin.linkappapi.blockqueue.QueueManager;
import com.easylinkin.linkappapi.blockqueue.constant.QueueKeyConstant;
import com.easylinkin.linkappapi.device.constant.DeviceConstant.DeviceWorkStatusEnum;
import com.easylinkin.linkappapi.device.entity.Device;
import com.easylinkin.linkappapi.device.entity.DeviceModel;
import com.easylinkin.linkappapi.device.mapper.DeviceMapper;
import com.easylinkin.linkappapi.device.mapper.DeviceModelMapper;
import com.easylinkin.linkappapi.device.service.DeviceService;
import com.easylinkin.linkappapi.device.service.ElectricyRecordsService;
import com.easylinkin.linkappapi.device.service.SprayRecordsService;
import com.easylinkin.linkappapi.device.service.WaterRecordsService;
import com.easylinkin.linkappapi.deviceattribute.entity.DeviceAttribute;
import com.easylinkin.linkappapi.deviceattribute.service.DeviceAttributeService;
import com.easylinkin.linkappapi.deviceattributestatus.entity.DeviceAttributeStatus;
import com.easylinkin.linkappapi.deviceattributestatus.service.DeviceAttributeStatusService;
import com.easylinkin.linkappapi.deviceunit.entity.DeviceUnit;
import com.easylinkin.linkappapi.deviceunit.service.DeviceUnitService;
import com.easylinkin.linkappapi.inspection.service.ElectricBoxRecordService;
import com.easylinkin.linkappapi.listener.AirMonitorListener;
import com.easylinkin.linkappapi.listener.DeviceAttributeStatusListener;
import com.easylinkin.linkappapi.listener.DeviceMonitorListener;
import com.easylinkin.linkappapi.lobar.service.GateService;
import com.easylinkin.linkappapi.machinery.service.ElevatorWorkRecordService;
import com.easylinkin.linkappapi.machinery.service.TowerCraneWorkRecordService;
import com.easylinkin.linkappapi.mechanical.service.CraneSafeMonitorService;
import com.easylinkin.linkappapi.openapi.dto.ApiDownResultDTO;
import com.easylinkin.linkappapi.openapi.dto.DatapushDTO;
import com.easylinkin.linkappapi.openapi.dto.OnlineState;
import com.easylinkin.linkappapi.openapi.dto.OnlineStateChangeDTO;
import com.easylinkin.linkappapi.openapi.service.DataPushService;
import com.easylinkin.linkappapi.openapi.service.SystemDockingService;
import com.easylinkin.linkappapi.positioning.service.PositionAlarmService;
import com.easylinkin.linkappapi.powerdistribution.service.DistributionCabinetStatusService;
import com.easylinkin.linkappapi.ruleengine.service.RuleEngineService;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.lucene.util.NamedThreadFactory;
import org.junit.Test;
import org.slf4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.*;

/**
 * <p>
 * </p>
 *
 * @author TongJie
 * @since 2020/11/7 14:14
 */
@Service
public class DataPushServiceImpl implements DataPushService {

	@Resource
	private DeviceAttributeStatusService deviceAttributeStatusService;
	@Resource
	private DeviceService deviceService;
	@Resource
	private DeviceAttributeService deviceAttributeService;
	@Resource
	private DeviceUnitService deviceUnitService;
	@Resource
	private RuleEngineService ruleEngineService;
	@Autowired
	private DistributionCabinetStatusService distributionCabinetStatusService;
	@Resource
	private DeviceMapper deviceMapper;
	@Resource
	private RedisUtil redisUtil;
	@Resource
	AirconditionDeviceRecordService airconditionDeviceRecordService;
	@Resource
	AirMonitorListener airMonitorListener;

	@Resource
	private PositionAlarmService positionAlarmService;

	@Resource
	DeviceAttributeStatusListener deviceAttributeStatusListener;
	@Resource
	private DeviceMonitorListener deviceMonitorListener;

	@Autowired
	private GateService gateService;

	@Autowired
	private SprayRecordsService sprayRecordsService;

	@Autowired
	private WaterRecordsService waterRecordsService;

	@Autowired
	private ElectricyRecordsService electricyRecordsService;

	@Autowired
	private TowerCraneWorkRecordService towerCraneWorkRecordService;

	@Autowired
	private ElectricBoxRecordService electricBoxRecordService;

	@Autowired
	private AiAlarmCountService aiAlarmCountService;

	@Autowired
	private ElevatorWorkRecordService elevatorWorkRecordService;

	@Autowired
	private SystemDockingService dockingService;

	@Autowired
	DeviceModelMapper deviceModelMapper;

	@Autowired
	private CraneSafeMonitorService craneSafeMonitorService;

	private static final ExecutorService executorService = Executors
			.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

	private static final Logger LOGGER = org.slf4j.LoggerFactory.getLogger(DataPushServiceImpl.class);

	ThreadPoolExecutor datapushThreadPool = new ThreadPoolExecutor(5, 25, 1L, TimeUnit.SECONDS,
			new LinkedBlockingQueue<>(2000), new NamedThreadFactory("datapush-thread"),
			new ThreadPoolExecutor.AbortPolicy());

	@Override
	public DeviceModel getDeviceModel(String deviceCode) {
		LambdaQueryWrapper<DeviceModel> deviceModelLambdaQueryWrapper = new LambdaQueryWrapper<>();
		deviceModelLambdaQueryWrapper.eq(DeviceModel::getCode, deviceCode).eq(DeviceModel::getRemoveStatus, 0);
		List deviceModelList = deviceModelMapper.selectList(deviceModelLambdaQueryWrapper);
		if(CollectionUtil.isNotEmpty(deviceModelList)) {
			return (DeviceModel) deviceModelList.get(0);
		}
		return new DeviceModel();
	}

	@Override
	public void onlineStateChange(List<OnlineState> onlineStates) {

		LOGGER.info("来自linkthings对onlineStateChange接口的调用，getData参数是:{}", JSON.toJSONString(onlineStates));
//        更新设备在离线状态
		for (OnlineState onlineState : onlineStates) {
			Device device = deviceService.findOneByDeviceCode(onlineState.getDevice_id());
			if (device == null) {
				LOGGER.info("查询aep设备查询不到：{}", onlineState);
				continue;
			}
			device.setModifyTime(new Date());
			device.setLastPushTime(new Date());
			device.setOnlineState(onlineState.getOnline_state());
			UpdateWrapper<Device> updateWrapper = new UpdateWrapper<>();
			updateWrapper.eq("code", onlineState.getDevice_id());
			deviceService.update(device, updateWrapper);
		}
		for (OnlineState onlineState : onlineStates) {
			if (ObjectUtils.isEmpty(onlineState.getDevice_id()) || ObjectUtils.isEmpty(onlineState.getOnline_state())) {
				continue;
			}
			// 更新设备状态
			Device deviceQuery = new Device();
			deviceQuery.setCode(onlineState.getDevice_id());
			List<Device> deviceList = deviceService.selectDevices(deviceQuery);
			for (Device device : deviceList) {
				sendDeviceMessage(device);
				ruleEngineService.triggerAssemblyLine(device, "onlineStateChange");
				// 铁路在离线告警
				craneSafeMonitorService.onlineStateChange(device, onlineState.getOnline_state());
			}
		}
	}

	@Override
	public void updateDeviceAttrStatusAndGenerateAlarm(final Device device, final DeviceUnit deviceUnit,
			final Object data, final Long timestamp) {
		LOGGER.info("updateDeviceAttrStatusAndGenerateAlarm方法被调用1，设备编码为：{},版本是：{}", device.getCode(),
				deviceUnit.getVersion());
		List<DeviceAttributeStatus> deviceAttributeStatusList = updateDeviceAttrStatus(device, deviceUnit, data,
				timestamp);
		sendDeviceMessage(device);
		// 20220104新增：设备工作状态为维保状态不产生告警
		Integer workStatus = device.getWorkStatus();
		if (workStatus != null && workStatus.equals(DeviceWorkStatusEnum.MAINTAIN.getType())) {
			return;
		}
		deviceGenerateAlarm(device, deviceUnit, data, deviceAttributeStatusList);
	}

	@Transactional(rollbackFor = Exception.class)
	public List<DeviceAttributeStatus> updateDeviceAttrStatus(final Device device, final DeviceUnit deviceUnit,
			final Object data, Long timestamp) {
		List<DeviceAttributeStatus> deviceAttributeStatusList = new ArrayList<>();

		QueryWrapper<DeviceAttribute> qw0 = new QueryWrapper();
		qw0.eq("device_unit_id", deviceUnit.getId());
		qw0.select("identifier", "name");
		List<DeviceAttribute> deviceAttributeList = deviceAttributeService.list(qw0);
		Map<String, String> attrsMap = new HashMap<>();
		for (DeviceAttribute deviceAttribute : deviceAttributeList) {
			attrsMap.put(deviceAttribute.getIdentifier(), deviceAttribute.getName());
		}

		Map<String, Object> mapObj = (Map) JSON.toJSON(data);
		for (Map.Entry<String, Object> entry : mapObj.entrySet()) {
			if (StringUtils.isBlank(entry.getKey())) {
				continue;
			}
			DeviceAttributeStatus deviceAttributeStatus0 = new DeviceAttributeStatus();

			deviceAttributeStatus0.setDeviceName(device.getName());
			deviceAttributeStatus0.setDeviceCode(device.getCode());
			deviceAttributeStatus0.setPropCode(entry.getKey());
			deviceAttributeStatus0.setPropName(attrsMap.get(entry.getKey()));
			deviceAttributeStatus0.setVersion(deviceUnit.getVersion());
			deviceAttributeStatus0.setTimestamp(timestamp);

			QueryWrapper<DeviceAttributeStatus> qw = new QueryWrapper<>();
			qw.eq("device_code", device.getCode());
			qw.eq("prop_code", entry.getKey());
			qw.eq("version", deviceUnit.getVersion());

			List<DeviceAttributeStatus> list = deviceAttributeStatusService.list(qw);
			if (list.isEmpty()) {
				Object obj = entry.getValue();
				deviceAttributeStatus0.setPropValue(String.valueOf(entry.getValue()));
				deviceAttributeStatus0.setCreateTime(new Date());
				deviceAttributeStatusService.saveOrUpdate2(deviceAttributeStatus0);
				LOGGER.info("datapush调用2后save已执行，参数是：{}", deviceAttributeStatus0);
				deviceAttributeStatusList.add(deviceAttributeStatus0);
				if (obj instanceof JSONObject) {
					deviceAttributeNodeSave(device, entry, deviceAttributeStatus0, obj, deviceAttributeStatusList, null,
							timestamp);
				}
				if (obj instanceof JSONArray) {
					JSONArray array = (JSONArray) obj;
					for (int i = 0; i < array.size(); i++) {
						if (array.get(i) instanceof JSONObject) {
							deviceAttributeNodeSave(device, entry, deviceAttributeStatus0, array.getJSONObject(i),
									deviceAttributeStatusList, i, timestamp);
						}
					}
				}
			} else {
				DeviceAttributeStatus deviceAttributeStatus = list.get(0);
				Object obj = entry.getValue();
				deviceAttributeStatus.setPropValue(String.valueOf(entry.getValue()));
				deviceAttributeStatus.setModifyTime(new Date());
				deviceAttributeStatus.setDeviceName(device.getName());
				deviceAttributeStatus.setPropName(attrsMap.get(entry.getKey()));
				deviceAttributeStatus.setTimestamp(timestamp);
				deviceAttributeStatusService.update2(deviceAttributeStatus);
				LOGGER.info("updateDeviceAttrStatus调用3后updateById已执行，deviceAttributeStatus参数是：{}",
						deviceAttributeStatus);

				deviceAttributeStatusList.add(deviceAttributeStatus);
				if (obj instanceof JSONObject) {
					deviceAttributeNodeUpdate(device, entry, deviceAttributeStatus, obj, deviceAttributeStatusList,
							null, timestamp);
				}
				if (obj instanceof JSONArray) {
					JSONArray array = (JSONArray) obj;
					for (int i = 0; i < array.size(); i++) {
						if (array.get(i) instanceof JSONObject) {
							deviceAttributeNodeUpdate(device, entry, deviceAttributeStatus, array.getJSONObject(i),
									deviceAttributeStatusList, i, timestamp);
						}
					}
				}
			}
		}
		return deviceAttributeStatusList;
	}

	private void sendDeviceMessage(Device device) {
		// 2021-07-23修改，避免被规则引擎给终止，推送设备数据
//        Device Device = deviceService.findOneByDeviceCode(device_id);
		// ws设备数据推送
		airMonitorListener.sendNotice(device);
		deviceAttributeStatusListener.sendNotice(device);
		deviceMonitorListener.sendNotice(device);
	}

	public void deviceGenerateAlarm(Device device1, DeviceUnit deviceUnit, Object data,
			List<DeviceAttributeStatus> deviceAttributeStatusList) {
		if (ObjectUtils.isEmpty(deviceAttributeStatusList)) {
			LOGGER.info("deviceGenerateAlarm方法被调用，设备编码为：{},对应设备属性状态为空！", device1.getCode());
			return;
		}
		// 设备是否触发报警
		if (ObjectUtils.isEmpty(deviceUnit)) {
			LOGGER.info("deviceGenerateAlarm方法被调用，设备编码为：{},查得型号为空", device1.getCode());
			return;
		}
		Device device = new Device();
		device.setName(device1.getName());
		device.setCode(device1.getCode());
		device.setDeviceUnitId(deviceUnit.getId());
		device.setSourceJson(JSONObject.toJSONString(data));
		device.setDeviceAttributeStatusList(deviceAttributeStatusList);
		try {
			distributionCabinetStatusService.checkGenerateDistributionCabinetStatus(device);
		} catch (Exception e) {
			LOGGER.error(e.getMessage(), e);
		}
		ruleEngineService.triggerAssemblyLine(device, data);
	}

	private void deviceAttributeNodeSave(final Device device, Map.Entry<String, Object> entry,
			DeviceAttributeStatus deviceAttributeStatus, Object obj,
			List<DeviceAttributeStatus> deviceAttributeStatusList, Integer arrayIndex, Long timestamp) {
		if (obj instanceof JSONObject) {
			Map<String, Object> map = (Map) obj;
			int j = 0;
			for (Map.Entry<String, Object> entryNode : map.entrySet()) {
				DeviceAttributeStatus deviceAttributeStatusNode = new DeviceAttributeStatus();
				deviceAttributeStatusNode.setDeviceName(device.getName());
				deviceAttributeStatusNode.setDeviceCode(device.getCode());
				deviceAttributeStatusNode.setPropName(entryNode.getKey());
				deviceAttributeStatusNode.setPropCode(entryNode.getKey());
				deviceAttributeStatusNode.setPropValue(String.valueOf(entryNode.getValue()));
				deviceAttributeStatusNode.setParentPropCode(deviceAttributeStatus.getPropCode());
				deviceAttributeStatusNode.setParentId(deviceAttributeStatus.getId());
				deviceAttributeStatusNode.setVersion(deviceAttributeStatus.getVersion());
				deviceAttributeStatusNode.setCreateTime(new Date());
				deviceAttributeStatusNode.setArrayIndex(arrayIndex);
				deviceAttributeStatusNode.setTimestamp(timestamp);
				deviceAttributeStatusService.saveOrUpdate2(deviceAttributeStatusNode);
				LOGGER.info("datapush调用4后save已执行，参数是：{}", deviceAttributeStatusNode);
				deviceAttributeStatusList.add(deviceAttributeStatusNode);
				if (entryNode.getValue() instanceof JSONObject) {
					deviceAttributeNodeSave(device, entryNode, deviceAttributeStatusNode, entryNode.getValue(),
							deviceAttributeStatusList, j, timestamp);
				}
				j++;
			}
		} else {
			deviceAttributeStatus.setDeviceName(device.getName());
			deviceAttributeStatus.setDeviceCode(device.getCode());
			deviceAttributeStatus.setPropName(entry.getKey());
			deviceAttributeStatus.setPropCode(entry.getKey());
			deviceAttributeStatus.setPropValue(String.valueOf(entry.getValue()));
			deviceAttributeStatus.setCreateTime(new Date());
			deviceAttributeStatus.setArrayIndex(arrayIndex);
			deviceAttributeStatus.setTimestamp(timestamp);
			deviceAttributeStatusService.saveOrUpdate2(deviceAttributeStatus);
			LOGGER.info("datapush调用5后save已执行，参数是：{}", deviceAttributeStatus);
			deviceAttributeStatusList.add(deviceAttributeStatus);
		}
	}

	private void deviceAttributeNodeUpdate(final Device device, Map.Entry<String, Object> entry,
			DeviceAttributeStatus deviceAttributeStatus, Object obj,
			List<DeviceAttributeStatus> deviceAttributeStatusList, Integer arrayIndex, Long timestamp) {
		if (obj instanceof JSONObject) {
			Map<String, Object> map = (Map) obj;
			int j = 0;
			for (Map.Entry<String, Object> entryNode : map.entrySet()) {
				QueryWrapper<DeviceAttributeStatus> qw = new QueryWrapper<DeviceAttributeStatus>();
				qw.eq("device_code", deviceAttributeStatus.getDeviceCode());
				qw.eq("prop_code", entryNode.getKey());
				qw.eq("parent_id", deviceAttributeStatus.getId());
				qw.eq("version", deviceAttributeStatus.getVersion());
				if (arrayIndex != null) {
					qw.eq("array_index", arrayIndex);
				}
				List<DeviceAttributeStatus> nodelist = deviceAttributeStatusService.list(qw);
				if (nodelist.isEmpty()) {
					DeviceAttributeStatus deviceAttributeStatusNode = new DeviceAttributeStatus();
					deviceAttributeStatusNode.setDeviceName(device.getName());
					deviceAttributeStatusNode.setDeviceCode(device.getCode());
					deviceAttributeStatusNode.setPropName(entryNode.getKey());
					deviceAttributeStatusNode.setPropCode(entryNode.getKey());
					deviceAttributeStatusNode.setPropValue(String.valueOf(entryNode.getValue()));
					deviceAttributeStatusNode.setParentPropCode(deviceAttributeStatus.getPropCode());
					deviceAttributeStatusNode.setParentId(deviceAttributeStatus.getId());
					deviceAttributeStatusNode.setVersion(deviceAttributeStatus.getVersion());
					deviceAttributeStatusNode.setCreateTime(new Date());
					deviceAttributeStatusNode.setArrayIndex(arrayIndex);
					deviceAttributeStatusNode.setTimestamp(timestamp);
					deviceAttributeStatusService.saveOrUpdate2(deviceAttributeStatusNode);
					LOGGER.info("datapush调用6后updateById已执行，参数是：{}", deviceAttributeStatusNode);
					deviceAttributeStatusList.add(deviceAttributeStatusNode);
					if (entryNode.getValue() instanceof JSONObject) {
						deviceAttributeNodeSave(device, entryNode, deviceAttributeStatusNode, entryNode.getValue(),
								deviceAttributeStatusList, j, timestamp);
					}
				} else {
					DeviceAttributeStatus deviceAttributeStatusNode = nodelist.get(0);
					deviceAttributeStatusNode.setDeviceName(device.getName());
					deviceAttributeStatusNode.setDeviceCode(device.getCode());
					deviceAttributeStatusNode.setPropName(entryNode.getKey());
					deviceAttributeStatusNode.setPropCode(entryNode.getKey());
					deviceAttributeStatusNode.setPropValue(String.valueOf(entryNode.getValue()));
					deviceAttributeStatusNode.setParentPropCode(deviceAttributeStatus.getPropCode());
					deviceAttributeStatusNode.setParentId(deviceAttributeStatus.getId());
					deviceAttributeStatusNode.setVersion(deviceAttributeStatus.getVersion());
					deviceAttributeStatusNode.setModifyTime(new Date());
					deviceAttributeStatusNode.setArrayIndex(arrayIndex);
					deviceAttributeStatusNode.setTimestamp(timestamp);
					deviceAttributeStatusService.update2(deviceAttributeStatusNode);
					LOGGER.info("datapush调用7后updateById已执行，参数是：{}", deviceAttributeStatusNode);
					deviceAttributeStatusList.add(deviceAttributeStatusNode);
					if (entryNode.getValue() instanceof JSONObject) {
						deviceAttributeNodeUpdate(device, entryNode, deviceAttributeStatusNode, entryNode.getValue(),
								deviceAttributeStatusList, j, timestamp);
					}
				}
				j++;
			}
		} else {
			QueryWrapper<DeviceAttributeStatus> qw = new QueryWrapper<DeviceAttributeStatus>();
			qw.eq("device_code", deviceAttributeStatus.getDeviceCode());
			qw.eq("prop_code", entry.getKey());
			qw.eq("parent_id", deviceAttributeStatus.getId());
			qw.eq("version", deviceAttributeStatus.getVersion());
			if (arrayIndex != null) {
				qw.eq("array_index", arrayIndex);
			}
			List<DeviceAttributeStatus> nodelist = deviceAttributeStatusService.list(qw);
			if (nodelist.isEmpty()) {
				deviceAttributeStatus.setDeviceName(device.getName());
				deviceAttributeStatus.setDeviceCode(device.getCode());
				deviceAttributeStatus.setPropName(entry.getKey());
				deviceAttributeStatus.setPropCode(entry.getKey());
				deviceAttributeStatus.setPropValue(String.valueOf(entry.getValue()));
				deviceAttributeStatus.setCreateTime(new Date());
				deviceAttributeStatus.setArrayIndex(arrayIndex);
				deviceAttributeStatus.setTimestamp(timestamp);
				deviceAttributeStatusService.saveOrUpdate2(deviceAttributeStatus);
				LOGGER.info("datapush调用8后save已执行， 参数是：{}", deviceAttributeStatus);
				deviceAttributeStatusList.add(deviceAttributeStatus);
			} else {
				deviceAttributeStatus = nodelist.get(0);
				deviceAttributeStatus.setDeviceName(device.getName());
				deviceAttributeStatus.setDeviceCode(device.getCode());
				deviceAttributeStatus.setPropName(entry.getKey());
				deviceAttributeStatus.setPropCode(entry.getKey());
				deviceAttributeStatus.setPropValue(String.valueOf(entry.getValue()));
				deviceAttributeStatus.setModifyTime(new Date());
				deviceAttributeStatus.setArrayIndex(arrayIndex);
				deviceAttributeStatus.setTimestamp(timestamp);
				deviceAttributeStatusService.update2(deviceAttributeStatus);
				LOGGER.info("datapush调用9后updateById已执行，参数是：{}", deviceAttributeStatus);
				deviceAttributeStatusList.add(deviceAttributeStatus);
			}
		}
	}

	/**
	 * 测试是否 满足重发条件
	 */
	@Test
	public void testNeedResend() {
//         String json = "{\"device_category\":\"空调控制器\",\"device_id\":\"YM12218H0150\",\"device_unit\":\"KTKZQ-NBAA-02-0036\",\"response\":\"{\\\"code\\\":0,\\\"message\\\":\\\"下发成功\\\",\\\"data\\\":{\\\"resultCode\\\":0},\\\"request_id\\\":\\\"193ac7e45c6d4d80b7508693951fb5cd\\\"}\",\"timestamp\":1631761153218,\"type\":2,\"uuid\":\"c863b08230fa494cb98cb2f66aee7f05\",\"version\":\"v1.3\"}";
//         String json = "{\"code\":1,\"device_category\":\"0829\",\"device_id\":\"351818114896012\",\"device_unit\":\"KTKZQ-NBAA-02-01-0036\",\"response\":\"{\\\"result\\\":{\\\"resultDetail\\\":{\\\"result\\\":0},\\\"resultCode\\\":\\\"COMPLETED\\\"},\\\"protocol\\\":\\\"lwm2m\\\",\\\"productId\\\":\\\"15087132\\\",\\\"messageType\\\":\\\"commandResponse\\\",\\\"tenantId\\\":\\\"2000016082\\\",\\\"deviceId\\\":\\\"8269b67a597d4c82b87aada507d076c7\\\",\\\"request_id\\\":\\\"9e2fd04bfc1947e390092050b75de8c3\\\",\\\"taskId\\\":127,\\\"timestamp\\\":1634119335422}\",\"timestamp\":1634119335515,\"type\":2,\"uuid\":\"43e1e2d389f145358c6c9a2d0b2a9633\",\"version\":\"1.0\"}";
//         String json = "{\"device_category\":\"0829\",\"device_id\":\"351818114896012\",\"device_unit\":\"KTKZQ-NBAA-02-01-0036\",\"response\":\"{\\\"protocol\\\":\\\"lwm2m\\\",\\\"productId\\\":\\\"15087132\\\",\\\"messageType\\\":\\\"commandResponse\\\",\\\"tenantId\\\":\\\"2000016082\\\",\\\"deviceId\\\":\\\"8269b67a597d4c82b87aada507d076c7\\\",\\\"request_id\\\":\\\"9e2fd04bfc1947e390092050b75de8c3\\\",\\\"taskId\\\":127,\\\"timestamp\\\":1634119335422}\",\"timestamp\":1634119335515,\"type\":2,\"uuid\":\"43e1e2d389f145358c6c9a2d0b2a9633\",\"version\":\"1.0\"}";
//         String json = "{\"code\":123,\"device_category\":\"0829\",\"device_id\":\"351818114896012\",\"device_unit\":\"KTKZQ-NBAA-02-01-0036\",\"response\":\"{\\\"protocol\\\":\\\"lwm2m\\\",\\\"productId\\\":\\\"15087132\\\",\\\"messageType\\\":\\\"commandResponse\\\",\\\"tenantId\\\":\\\"2000016082\\\",\\\"deviceId\\\":\\\"8269b67a597d4c82b87aada507d076c7\\\",\\\"request_id\\\":\\\"90354b4ada2247cb93091a4a621456f0\\\",\\\"taskId\\\":135,\\\"timestamp\\\":1634121421874}\",\"timestamp\":1634121422051,\"type\":2,\"uuid\":\"77d60321df134a4bafefb9b4f0eb8d10\",\"version\":\"1.0\"}";
		String json = "{\"device_category\":\"0829\",\"device_id\":\"351818114896012\",\"device_unit\":\"KTKZQ-NBAA-02-01-0036\",\"response\":\"{\\\"code\\\":1,\\\"protocol\\\":\\\"lwm2m\\\",\\\"productId\\\":\\\"15087132\\\",\\\"messageType\\\":\\\"commandResponse\\\",\\\"tenantId\\\":\\\"2000016082\\\",\\\"deviceId\\\":\\\"8269b67a597d4c82b87aada507d076c7\\\",\\\"request_id\\\":\\\"90354b4ada2247cb93091a4a621456f0\\\",\\\"taskId\\\":135,\\\"timestamp\\\":1634121421874}\",\"timestamp\":1634121422051,\"type\":2,\"uuid\":\"77d60321df134a4bafefb9b4f0eb8d10\",\"version\":\"1.0\"}";
		JSONObject object = JSONObject.parseObject(json);
		if (object.containsKey("notify_type")) {
			System.out.println("无需重发1");
		} else {
			DatapushDTO datapushDTO = object.toJavaObject(DatapushDTO.class);
			if (Objects.equals(datapushDTO.getType(), 2)) {
				ApiDownResultDTO dataPushResponse = null;
				if (JSONUtil.isJsonObj(datapushDTO.getResponse())) {
					ApiDownResultDTO resp = JSON.parseObject(datapushDTO.getResponse(), ApiDownResultDTO.class);
					dataPushResponse = resp;
				}

				boolean needToResend = false;
				// 判断是否重发
				// 先判断内层,不存在code或者code不为0则为重发
				if (dataPushResponse == null || StringUtils.isEmpty(dataPushResponse.getCode())
						|| !"0".equalsIgnoreCase(dataPushResponse.getCode())) {
					needToResend = true;
				}
				// 再进行判断外层是否存在1,如果为1就不重发。如果有外层code，用外层兜底
				if (StringUtils.isNotEmpty(datapushDTO.getCode())) {
					needToResend = !"1".equals(datapushDTO.getCode());
				}
//                 boolean needToResend = "2".equals(datapushDTO.getCode()) || dataPushResponse == null || (dataPushResponse.getCode() != null && !"0".equalsIgnoreCase(dataPushResponse.getCode()));
//                 needToResend = !"1".equals(datapushDTO.getCode()) && needToResend;
//                 if (needToResend) {
//                     LOGGER.info("需要再重发,response:{}", dataPushResponse);
//                 }

				if (needToResend) {
					LOGGER.info("需要再重发,response:{}", dataPushResponse);
				}
			}
		}
	}

	/**
	 * 需使用线程池，误删。由于linkthings datapush推送数据，5秒未返回做了熔断。若返回超时将触发熔断。
	 * outputStream.write() 是伪异步，多个请求过来，发现均在一个线程中去处理的
	 *
	 * @param json
	 */
	@Override
	public void dataPush(String json) {
//        监控打印线程状态
		int queueSize = datapushThreadPool.getQueue().size();
		int activeCount = datapushThreadPool.getActiveCount();
		long completedTaskCount = datapushThreadPool.getCompletedTaskCount();
		long taskCount = datapushThreadPool.getTaskCount();
		LOGGER.info("datapush-当前活动线程数：" + activeCount + ",当前排队线程数：" + queueSize + ",执行完成线程数：" + completedTaskCount
				+ ",总线程数：" + taskCount);
		datapushThreadPool.execute(() -> {
			JSONObject object = JSONObject.parseObject(json);
//             如果是在离线
			if (object.containsKey("notify_type")) {
				LOGGER.info("dataPush被调用,类型是离线，datapush流水报文是:{}", json);
				OnlineStateChangeDTO onlineStateChangeDTO = JSON.parseObject(json, OnlineStateChangeDTO.class);
				Assert.notEmpty(onlineStateChangeDTO.getData(), "设备不能为空");
				onlineStateChange(onlineStateChangeDTO.getData());
			} else {
				DatapushDTO datapushDTO = object.toJavaObject(DatapushDTO.class);
				dataPushCommonFlow(datapushDTO);
			}
		});
	}

	/**
	 * 覆盖数据推送对象，将字符串长度大于2000的重新复制，忽略处理。解決传的报文长度太长而且用不到，打印在日志中白占用空间
	 *
	 * @param data
	 */
	private void coverDataPushDTOData(Map data) {
		for (Object key : data.keySet()) {
			Object d1 = data.get(key);
			if (d1 instanceof Map) {
				coverDataPushDTOData((Map) d1);
			} else if (d1 instanceof String) {
				if (((String) d1).length() > 2000) {
					data.put(key, "length over 2000 ,so set ignored");
				}
			}
		}
	}

	/**
	 * 普通流水的逻辑处理，非离线
	 *
	 * @param datapushDTO 普通流水的逻辑处理
	 */
	private void dataPushCommonFlow(DatapushDTO datapushDTO) {
		Assert.notNull(datapushDTO, "datapushDTO不能为空。");
		if (datapushDTO.getData() != null) {
			coverDataPushDTOData(datapushDTO.getData());
		}
		// 更新 设备推送时间取device设置中的,得区分开，有的情形的推送不更新 lastpushtime(如荆州油烟定时计算，并推送数据)
		LOGGER.info("dataPushCommonFlow被调用，参数是：{}", JSONObject.toJSONString(datapushDTO));
		Device device = deviceService.findOneByDeviceCode(datapushDTO.getDevice_id());
		if (device == null) {
			LOGGER.warn("查找设备为空,参数是:{}", JSONObject.toJSONString(datapushDTO));
			return;
		}
		device.setLastPushTime(new Date()).setOnlineState(1);
		deviceService.updateByCode(device, datapushDTO.getDevice_id());

		// type为2 是属于应答流水类型。 非2是正常流水
		if (Objects.equals(datapushDTO.getType(), 2)) {
			ApiDownResultDTO dataPushResponse = null;
			if (JSONUtil.isJsonObj(datapushDTO.getResponse())) {
				dataPushResponse = JSON.parseObject(datapushDTO.getResponse(), ApiDownResultDTO.class);
			}
			boolean needToResend = false;
			// 判断是否重发
			// 先判断内层,不存在code或者code不为0则为重发
			if (dataPushResponse == null || StringUtils.isEmpty(dataPushResponse.getCode())
					|| !"0".equalsIgnoreCase(dataPushResponse.getCode())) {
				needToResend = true;
			}
			// 再进行判断外层是否存在1,如果为1就不重发。如果有外层code，用外层兜底
			if (StringUtils.isNotEmpty(datapushDTO.getCode())) {
				needToResend = !"1".equals(datapushDTO.getCode());
			}
			if (needToResend) {
				LOGGER.info("需要再重发,response:{}", dataPushResponse);
			} else {
				assert dataPushResponse != null;
				String requestId = dataPushResponse.getRequestId();
				if (redisUtil.hasKey(requestId + "_down_param")) {
					redisUtil.del(requestId + "_down_param");
					LOGGER.info("不需要再重发,response:{}", dataPushResponse);
				}
			}
		} else {
//                     modify by tongjie 20210722，还原之前代码。 需求 - ack中带有属性值, 需写入设备属性状态表。
//                     linkthings会分别推2条，一条type=2的不携带设备属性值的，另一条携带设备属性值的流水。
			Assert.notNull(datapushDTO.getDevice_id(), "设备ID不能为空");
			Assert.notNull(datapushDTO.getDevice_unit(), "设备型号不能为空");
			Assert.notNull(datapushDTO.getData(), "报文不能为空");
			QueryWrapper<DeviceUnit> qw = new QueryWrapper<>();
			qw.eq("code", datapushDTO.getDevice_unit());
			qw.eq("version", datapushDTO.getVersion());
			List<DeviceUnit> list = deviceUnitService.list(qw);
			Assert.notEmpty(list, "查询得设备型号为空！");
			DeviceUnit deviceUnit0 = list.get(0);
			updateDeviceAttrStatusAndGenerateAlarm(device, deviceUnit0, datapushDTO.getData(),
					datapushDTO.getTimestamp());
			// 进入扩展逻辑(notify_type类型的推送数据data是Arrays结构，JSONObject无法解析)，示例：{"data":[{"device_id":"004A770124089745","online_state":1}],"device_id":"004A770124089745","notify_type":"online_state_change","timestamp":1631017494675,"uuid":"43efaaaf3955473aab6254d6cca5c001"}
			functionExtension(datapushDTO);
		}

	}

	@Resource
	private QueueManager queueManager;
	/**
	 * 定制化功能扩展逻辑 业务扩展时开启的事务需要：Propagation.REQUIRES_NEW 避免影响到设备属性状态变更
	 * 20230815直接推送到各阻塞队列 由队列自己处理 避免影响流水主线程的处理效率
	 * @param datapushDTO
	 */
	private void functionExtension(DatapushDTO datapushDTO) {

		// 1.公共流水校验
		if(datapushDTO == null){
			return;
		}
		String deviceId = datapushDTO.getDevice_id();
		if(org.springframework.util.StringUtils.isEmpty(deviceId)){
			return;
		}

		JSONObject jsonObject = datapushDTO.getData();
		if(jsonObject == null){
			return;
		}

		Device query = new Device();
		query.setCode(deviceId);
		List<Device> deviceList = deviceService.selectDevices(query);
		if (CollectionUtil.isEmpty(deviceList)) {
			return;
		}

		// 处理人脸识别逻辑 施工云没有的功能
//		executorService.execute(new FaceRecognitionTask(datapushDTO));

		// 空调分户计量-耗能统计 施工云没有的功能
//		airconditionDeviceRecordService.separateMetering(datapushDTO);

		// 电子围栏告警功能
//		positionAlarmService.checkPostionAlarm(datapushDTO);
		queueManager.enqueue(QueueKeyConstant.ELECTRONIC_FENCE_ALARM_FUNCTION,datapushDTO);

		// 闸机流水处理
//		gateService.datapushHandler(datapushDTO);
		queueManager.enqueue(QueueKeyConstant.GATE_FLOW_PROCESSING,datapushDTO);



		// 以下队列需要用到最新的设备数据
		datapushDTO.setDevice_data_latest(deviceList.get(0));
		// ai相机预警流水统计
		aiAlarmCountService.datapushHandler(datapushDTO);
		queueManager.enqueue(QueueKeyConstant.AI_CAMERA_WARNING_FLOW_STATISTICS,datapushDTO);
		// 喷淋流水处理
//		sprayRecordsService.datapushHandler(datapushDTO);
		queueManager.enqueue(QueueKeyConstant.SPRAY_FLOW_PROCESSING,datapushDTO);

		// 用水流水处理
//		waterRecordsService.datapushHandler(datapushDTO);
		queueManager.enqueue(QueueKeyConstant.WATER_FLOW_PROCESSING,datapushDTO);

		// 用电流水处理
//		electricyRecordsService.datapushHandler(datapushDTO);
		queueManager.enqueue(QueueKeyConstant.ELECTRICITY_FLOW_PROCESSING,datapushDTO);

		// 塔机流水处理
//			towerCraneWorkRecordService.datapushHandler(datapushDTO);
		queueManager.enqueue(QueueKeyConstant.TOWER_CRANE_FLOW_PROCESSING,datapushDTO);

		// 配电箱流水处理
//			electricBoxRecordService.datapushHandler(datapushDTO);
		queueManager.enqueue(QueueKeyConstant.ELECTRRIC_BOX_FLOW_PROCESSING,datapushDTO);

		// 升降机流水处理
//			elevatorWorkRecordService.datapushHandler(datapushDTO);
		queueManager.enqueue(QueueKeyConstant.LIFT_FLOW_PROCESSING,datapushDTO);

		// 定位器流水处理
		queueManager.enqueue(QueueKeyConstant.LOCATION_PROCESSING, datapushDTO);


		// 扬尘、塔机、升降机设备数据上传到省厅系统
//			dockingService.pushToSystemDockHandler(datapushDTO);
		queueManager.enqueue(QueueKeyConstant.DATA_UPLOAD_TO_PROVINCIAL_HALL_SYSTEM,datapushDTO);

		//断路器流水处理
		queueManager.enqueue(QueueKeyConstant.RAIL_CIRCUIT_GIVEALARM_FAULT_PROCESSING,datapushDTO);

		// 吊车流水处理
//		craneSafeMonitorService.datapushHandler(datapushDTO);
		queueManager.enqueue(QueueKeyConstant.RAIL_CRANE_SAFETY_MONITORING, datapushDTO);

		//电器火灾流水处理
		queueManager.enqueue(QueueKeyConstant.RAIL_CIRCUIT_DQHZ_PROCESSING,datapushDTO);


		// 北斗定位流水处理
		queueManager.enqueue(QueueKeyConstant.BEIDOU_LOCATION, datapushDTO);


		//AI综合安防行为流水记录处理
		//queueManager.enqueue(QueueKeyConstant.RAIL_AI_HOST_RECORD_PROCESSING, datapushDTO);

	}
}
